package window

import bean.SensorReading
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, OutputTag, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingAlignedProcessingTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/13 23:32
  *
  *
  *
  */
object WindowTest {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //设置事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)
    val socketStream = env.socketTextStream("192.168.88.106", 7777)

    //侧输出流
    val lateTag = new OutputTag[(String, Long, Double)]("late")

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = socketStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
//      .assignAscendingTimestamps(_.timestamp*1000L)  //升序数据
      //乱序数据设置watermark
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(3)) {
      override def extractTimestamp(t: SensorReading): Long = t.timestamp*1000L
    })
    //    dataStream.print()

    val resultStream: DataStream[(String, Long, Double)] = dataStream
      .map(data => (data.id, data.timestamp, data.temperature))
      .keyBy(data => data._1) //.keyBy(_._1)
      //      .window(TumblingEventTimeWindows.of(Time.days(1),Time.hours(-8)))  //滚动窗口 00:00~24:00 东八区
      //      .window(TumblingEventTimeWindows.of(Time.seconds(15)))  //滚动窗口
      //      .window(SlidingAlignedProcessingTimeWindows.of(Time.seconds(15),Time.seconds(3)))  //滑动窗口
      //      .window(EventTimeSessionWindows.withGap(Time.seconds(10))) //会话窗口
      //      .timeWindow(Time.seconds(15),Time.seconds(3)) //滑动窗口
      //      .timeWindow(Time.seconds(15))  //滚动窗口
      //      .countWindow(10) //滚动窗口
      //      .countWindow(10,3) //滑动窗口

      .timeWindow(Time.seconds(15))
      .allowedLateness(Time.minutes(1))
      .sideOutputLateData(lateTag)

      //      .minBy(1)
      .reduce((curRes, newData) => (curRes._1, newData._2, curRes._3.min(newData._3)))

    resultStream.getSideOutput(lateTag).print("late")

    resultStream.print("window")


    env.execute("window test")
  }

  //自定义ReduceFunction
  class MyReducer extends ReduceFunction[SensorReading] {
    override def reduce(t: SensorReading, t1: SensorReading): SensorReading = {
      SensorReading(t.id, t1.timestamp, t.temperature.min(t1.temperature))
    }
  }

}
